package com.sh.data.engine.domain.integration.datasource.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.EnumUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.druid.DbType;
import com.alibaba.druid.sql.ast.statement.SQLCreateTableStatement;
import com.alibaba.druid.sql.parser.SQLParserUtils;
import com.alibaba.druid.sql.parser.SQLStatementParser;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.collect.Lists;
import com.sh.data.engine.common.constants.Constants.RedisKey;
import com.sh.data.engine.common.constants.HttpStatus;
import com.sh.data.engine.common.constants.code.BizCodes;
import com.sh.data.engine.common.enumDefinition.DSType;
import com.sh.data.engine.common.exception.BusinessException;
import com.sh.data.engine.common.util.DataUtil;
import com.sh.data.engine.common.util.RSAUtil;
import com.sh.data.engine.common.util.ServletUtils;
import com.sh.data.engine.domain.authority.service.DataAuthService;
import com.sh.data.engine.domain.base.model.PageResult;
import com.sh.data.engine.domain.base.model.UserContext;
import com.sh.data.engine.domain.base.model.enums.CreateTableCaseEnum;
import com.sh.data.engine.domain.base.model.enums.Database;
import com.sh.data.engine.domain.base.model.enums.RightEnum;
import com.sh.data.engine.domain.dataasset.model.domain.TableInfoBizDomain;
import com.sh.data.engine.domain.dataasset.model.param.QueryTableParam;
import com.sh.data.engine.domain.dataasset.service.TableService;
import com.sh.data.engine.domain.integration.datasource.model.domain.*;
import com.sh.data.engine.domain.integration.datasource.model.mapper.*;
import com.sh.data.engine.domain.integration.datasource.model.param.*;
import com.sh.data.engine.domain.integration.datasource.service.DataSourceService;
import com.sh.data.engine.domain.shims.DbManagerFactory;
import com.sh.data.engine.domain.shims.db.BaseDbManager;
import com.sh.data.engine.domain.shims.db.DbOptions;
import com.sh.data.engine.domain.shims.db.model.FieldInfoDomain;
import com.sh.data.engine.domain.shims.db.model.PreviewDataDomain;
import com.sh.data.engine.domain.shims.db.model.TableInfoDomain;
import com.sh.data.engine.domain.shims.dgraph.util.DgraphUtil;
import com.sh.data.engine.domain.shims.hbase.manager.HBaseManager;
import com.sh.data.engine.domain.shims.hbase.util.HBaseUtil;
import com.sh.data.engine.domain.shims.hdfs.util.HdfsUtil;
import com.sh.data.engine.domain.shims.influxdb.manager.InfluxDBManager;
import com.sh.data.engine.domain.shims.influxdb.option.InfluxDbOptions;
import com.sh.data.engine.domain.shims.iotdb.domain.DeviceDomain;
import com.sh.data.engine.domain.shims.iotdb.manager.IoTDBSessionManager;
import com.sh.data.engine.domain.shims.iotdb.option.IoTDBOptions;
import com.sh.data.engine.domain.shims.kafka.util.KafkaUtil;
import com.sh.data.engine.domain.shims.mongo.MongoOptions;
import com.sh.data.engine.domain.shims.mongo.manager.MongoManager;
import com.sh.data.engine.domain.shims.mqtt.util.MQTTUtil;
import com.sh.data.engine.domain.shims.opentsdb.util.OpenTSDBUtil;
import com.sh.data.engine.domain.shims.sql.ddl.CreateTableParam;
import com.sh.data.engine.domain.shims.sql.ddl.DDL;
import com.sh.data.engine.domain.util.ConvertUtil;
import com.sh.data.engine.domain.util.DBUtil;
import com.sh.data.engine.domain.util.SqlParserUtil;
import com.sh.data.engine.domain.workspace.index.service.ProjectService;
import com.sh.data.engine.domain.workspace.manager.model.domain.UserDomain;
import com.sh.data.engine.domain.workspace.manager.service.UserService;
import com.sh.data.engine.repository.dao.integration.datasource.DataSourceMapper;
import com.sh.data.engine.repository.dao.integration.datasource.entity.DataSourceEntity;
import com.sh.data.engine.repository.dao.integration.datasource.entity.DataSourceEntity.HDFSConfig;
import com.sh.data.engine.repository.dao.integration.datasource.entity.DataSourceEntity.HiveConfig;
import com.sh.data.engine.repository.dao.integration.datasource.entity.DataSourceEntity.MongoConfig;
import com.sh.data.engine.repository.dao.integration.datasource.entity.DataSourceEntity.RdbmsConfig;
import com.sh.data.engine.repository.dao.workspace.index.entity.ProjectEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.net.URLEncoder;
import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * 数据源管理服务
 *
 * @author: mengzheng.mhc
 * @date: 2024/5/29 17:21
 */
@Slf4j
@Service
public class DataSourceServiceImpl extends ServiceImpl<DataSourceMapper, DataSourceEntity> implements DataSourceService {

    @Autowired
    private ProjectService projectService;

    @Autowired
    private UserService userService;

    @Autowired
    private TableService tableService;

    @Autowired
    private DataAuthService dataAuthService;

    @Autowired
    private DataSourceEntityMapper dataSourceEntityMapper;

    @Autowired
    private RdbmsConfigMapper rdbmsConfigMapper;

    @Autowired
    private HiveConfigMapper hiveConfigMapper;

    @Autowired
    private MongoConfigMapper mongoConfigMapper;

    @Autowired
    private HDFSConfigMapper hdfsConfigMapper;

    @Autowired
    private IoTDBOptionsMapper ioTDBOptionsMapper;

    @Autowired
    private DataSourceDomainMapper dataSourceDomainMapper;

    @Autowired
    private DataSourceMultiDomainMapper dataSourceMultiDomainMapper;

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    private static final Pattern pattern = Pattern.compile("'(.*)'");

    @Override
    public PageResult<DataSourceDomain> getDataSourcePageList(DataSourceQueryParam param) {
        var lambdaQueryWrapper =
            new LambdaQueryWrapper<DataSourceEntity>()
                .eq(StringUtils.isNotBlank(param.getDsType()), DataSourceEntity::getDsType, param.getDsType())
                .like(StringUtils.isNotBlank(param.getDsName()), DataSourceEntity::getDsName, param.getDsName())
                .eq(Objects.nonNull(param.getStatus()), DataSourceEntity::getStatus, param.getStatus())
                .eq(Objects.nonNull(param.getProjectId()), DataSourceEntity::getProjectId, param.getProjectId())
                .orderByDesc(DataSourceEntity::getUpdateTime);

        IPage<DataSourceEntity> dataSourcePage = this.page(new Page<>(param.getPageNum(), param.getPageSize()), lambdaQueryWrapper);

        PageResult<DataSourceEntity> dataSourcePageResult = PageResult.convert(dataSourcePage);

        return ConvertUtil.convertPageInfo(dataSourcePageResult, DataSourceDomain.class);
    }

    @Override
    public DataSourceDetailDomain getDataSourceDetailById(Long id, boolean encrypt) {
        DataSourceEntity dataSourceEntity = this.baseMapper.selectById(id);
        if (Objects.isNull(dataSourceEntity)) {
            throw new BusinessException(BizCodes.DS_NOT_EXISTS_ERROR, HttpStatus.BAD_REQUEST);
        }
        DataSourceDetailDomain dataSourceDetailDomain = dataSourceDomainMapper.map(dataSourceEntity);
        String dsType = dataSourceEntity.getDsType();

        if (DSType.isRdbms(dsType)
            || DSType.IoTDB.toString().equals(dsType)
            || DSType.InfluxDB.toString().equals(dsType)) {
            RdbmsConfig rdbmsConfig = dataSourceEntity.getRdbmsConfig();
            String rdbmsPassword = rdbmsConfig.getPassword();
            if (StringUtils.isNotBlank(rdbmsPassword)) {
                try {
                    if (encrypt) {
                        rdbmsPassword = URLEncoder.encode(rdbmsPassword, "UTF-8");
                        dataSourceDetailDomain.setPassword(RSAUtil.encryptByPublicKey(rdbmsPassword));
                    } else {
                        dataSourceDetailDomain.setPassword(rdbmsPassword);
                    }
                } catch (Exception e) {
                    log.error(e.getMessage());
                    throw new BusinessException(BizCodes.ENCRYPT_PASSWORD_FAILED, HttpStatus.BAD_REQUEST);
                }
            }
            DataSourceDetailDomain.RdbmsConfig rdbmsConfigDetail = rdbmsConfigMapper.entityToDomain(rdbmsConfig);
            dataSourceDetailDomain.setRdbmsConfig(rdbmsConfigDetail);
            dataSourceDetailDomain.setDbName(rdbmsConfig.getDbName());
            dataSourceDetailDomain.setUsername(rdbmsConfig.getUsername());
        } else if (DSType.Hive.name().equalsIgnoreCase(dsType)) {
            HiveConfig hiveConfig = dataSourceEntity.getHiveConfig();
            String hivePassword = hiveConfig.getPassword();
            if (StringUtils.isNotBlank(hivePassword)) {
                try {
                    if (encrypt) {
                        hivePassword = URLEncoder.encode(hivePassword, "UTF-8");
                        dataSourceDetailDomain.setPassword(RSAUtil.encryptByPublicKey(hivePassword));
                    } else {
                        dataSourceDetailDomain.setPassword(hivePassword);
                    }
                } catch (Exception e) {
                    log.error(e.getMessage());
                    throw new BusinessException(BizCodes.ENCRYPT_PASSWORD_FAILED, HttpStatus.BAD_REQUEST);
                }
            }
            dataSourceDetailDomain.setUsername(hiveConfig.getUsername());
            DataSourceDetailDomain.HiveConfig hiveConfigDetail = hiveConfigMapper.entityToDomain(hiveConfig);
            dataSourceDetailDomain.setHiveConfig(hiveConfigDetail);
            dataSourceDetailDomain.setDbName(hiveConfig.getDbName());
            dataSourceDetailDomain.setUsername(hiveConfig.getUsername());
        } else if (DSType.Mongodb.name().equalsIgnoreCase(dsType)) {
            MongoConfig mongoConfig = dataSourceEntity.getMongoConfig();
            String password = mongoConfig.getPassword();
            if (StringUtils.isNotBlank(password)) {
                try {
                    if (encrypt) {
                        password = URLEncoder.encode(password, "UTF-8");
                        dataSourceDetailDomain.setPassword(RSAUtil.encryptByPublicKey(password));
                    } else {
                        dataSourceDetailDomain.setPassword(password);
                    }
                } catch (Exception e) {
                    log.error(e.getMessage());
                    throw new BusinessException(BizCodes.ENCRYPT_PASSWORD_FAILED, HttpStatus.BAD_REQUEST);
                }
            }

            DataSourceDetailDomain.MongoConfig mongoConfigDetail = mongoConfigMapper.entityToDomain(mongoConfig);
            dataSourceDetailDomain.setMongoConfig(mongoConfigDetail);
            dataSourceDetailDomain.setDbName(mongoConfig.getDbName());
            dataSourceDetailDomain.setUsername(mongoConfig.getUsername());
//        } else if (DSType.MaxCompute.name().equalsIgnoreCase(dsType)) {
//            MaxComputeConfig maxComputeConfig = dataSourceEntity.getMaxComputeConfig();
//            DataSourceDetailDomain.MaxComputeConfig maxComputeDetail =
//                ConvertUtil.copyProperties(
//                    maxComputeConfig, DataSourceDetailDomain.MaxComputeConfig.class);
//            dataSourceDetailDomain.setMaxComputeConfig(maxComputeDetail);
        } else if (DSType.HDFS.name().equalsIgnoreCase(dsType)) {
            HDFSConfig hdfsConfig = dataSourceEntity.getHdfsConfig();
            DataSourceDetailDomain.HDFSConfig hdfsConfigDetail = hdfsConfigMapper.entityToDomain(hdfsConfig);
            dataSourceDetailDomain.setHdfsConfig(hdfsConfigDetail);
        }

        return dataSourceDetailDomain;
    }

    /**
     * @param param 数据源保存参数
     * @return
     */
    @Override
    public Long saveDataSource(DataSourceSaveParam param) {
        boolean flag = true;
        String dsLink = null;
        String username;
        String password;
        if (Objects.isNull(param.getId())) {
            flag = false;
        }
        String dsType = param.getDsType();
        if (StringUtils.isBlank(dsType)) {
            throw new BusinessException(BizCodes.DS_TYPE_INVALID_ERROR, HttpStatus.BAD_REQUEST);
        }
        if (doesDsNameExist(param)) {
            throw new BusinessException(BizCodes.DS_NAME_ALREADY_EXISTS_ERROR, HttpStatus.BAD_REQUEST);
        }

        DataSourceEntity dataSourceEntity = dataSourceEntityMapper.saveParamToEntity(param);
        if (param.getId() != null) {
            DataSourceEntity dataSource = this.getBaseMapper().selectById(param.getId());
            if (dataSource == null) {
                throw new BusinessException(BizCodes.DS_NOT_EXISTS_ERROR, HttpStatus.BAD_REQUEST);
            }
        }
        int status = 1;
        if (DSType.isRdbms((dsType))) {
            DataSourceRdbmsSaveParam rdbmsSaveDomain = (DataSourceRdbmsSaveParam) param;
            username = rdbmsSaveDomain.getUsername();
            password = rdbmsSaveDomain.getPassword();
            dsLink = rdbmsSaveDomain.getDsLink();

            String version = param.getVersion();
            Integer statusParam = param.getStatus();
            RdbmsConfig rdbmsConfig = rdbmsConfigMapper.paramToConfig(rdbmsSaveDomain);
            if (StringUtils.isBlank(version)) {
                BaseDbManager dbManager = getDbManager(dsType, username, password, dsLink);
                String dbVersion = null;
                try {
                    dbVersion = dbManager.getDbVersion();
                } catch (Exception e) {
                    log.error(e.getMessage());
                }
                try {
                    dbManager.testConnection();
                } catch (SQLException e) {
                    status = 0;
                    log.error(e.getMessage());
                }
                rdbmsConfig.setVersion(dbVersion);
                dataSourceEntity.setStatus(status);
            } else {
                rdbmsConfig.setVersion(version);
                dataSourceEntity.setStatus(statusParam);
            }
            dataSourceEntity.setDsLink(dsLink);
            dataSourceEntity.setRdbmsConfig(rdbmsConfig);

        } else if (dsType.equalsIgnoreCase(DSType.Hive.name())) {
            DataSourceHiveSaveParam dataSourceHiveParam = (DataSourceHiveSaveParam) param;
            username = dataSourceHiveParam.getUsername();
            password = dataSourceHiveParam.getPassword();
            dsLink = dataSourceHiveParam.getDsLink();
            String hiveSiteAddress = dataSourceHiveParam.getHiveSiteAddress();
            HiveConfig hiveConfig = hiveConfigMapper.paramToConfig(dataSourceHiveParam);
            Integer paramStatus = param.getStatus();
            String version = param.getVersion();
            String dbVersion = null;
            if (StringUtils.isBlank(version) && Objects.isNull(paramStatus)) {
                try {
                    BaseDbManager dbManager =
                        getDbManager(dsType, username, password, dsLink, hiveSiteAddress);
                    dbVersion = dbManager.getDbVersion();
                    dbManager.testConnection();
                } catch (Exception e) {
                    status = 0;
                    log.error(e.getMessage());
                }
                hiveConfig.setVersion(dbVersion);
                dataSourceEntity.setStatus(status);
            } else {
                hiveConfig.setVersion(version);
                dataSourceEntity.setStatus(paramStatus);
            }
            dataSourceEntity.setDsLink(dsLink);
            dataSourceEntity.setHiveConfig(hiveConfig);
        } else if (dsType.equalsIgnoreCase(DSType.Kafka.name())) {
            dsLink = param.getDsLink();
            try {
                KafkaUtil.testConnection(dsLink, 3);
            } catch (Exception e) {
                status = 0;
                log.error(e.getMessage());
            }
            dataSourceEntity.setStatus(status);
        }
        // mqtt
        else if (dsType.equalsIgnoreCase(DSType.MQTT.name())) {
            dsLink = param.getDsLink();
            try {
                MQTTUtil.testConnection(dsLink, null);
            } catch (Exception e) {
                status = 0;
                log.error(e.getMessage());
            }
            dataSourceEntity.setStatus(status);
        } else if (dsType.equalsIgnoreCase(DSType.Mongodb.name())) {
            DataSourceMongoSaveParam dataSourceMongodbParam = (DataSourceMongoSaveParam) param;
            MongoOptions mongoOptions = new MongoOptions();
            mongoOptions.setAddressList(Collections.singletonList(dataSourceMongodbParam.getDsLink()));
            mongoOptions.setAuthDbName(dataSourceMongodbParam.getAuthDbName());
            mongoOptions.setDbName(dataSourceMongodbParam.getDbName());
            mongoOptions.setUsername(dataSourceMongodbParam.getUsername());
            mongoOptions.setPassword(dataSourceMongodbParam.getPassword());
            MongoManager mongoManager = new MongoManager(mongoOptions);
            try {
                mongoManager.testConnection();
            } catch (Exception e) {
                status = 0;
                log.error(e.getMessage());
            }
            MongoConfig mongoConfig = mongoConfigMapper.paramToConfig(dataSourceMongodbParam);
            dataSourceEntity.setStatus(status);
            dataSourceEntity.setMongoConfig(mongoConfig);
        } else if (dsType.equalsIgnoreCase(DSType.HBase.name())) {
            try {
                dsLink = param.getDsLink();
                HBaseUtil.testConnection(dsLink);
            } catch (Exception e) {
                status = 0;
                log.error(e.getMessage());
            }
            dataSourceEntity.setStatus(status);
        } else if (dsType.equalsIgnoreCase(DSType.HDFS.name())) {
            DataSourceHDFSSaveParam dataSourceHDFSParam = (DataSourceHDFSSaveParam) param;
            HDFSConfig hdfsConfig = hdfsConfigMapper.paramToConfig(dataSourceHDFSParam);
            String hdfsUser = hdfsConfig.getHdfsUser();
            dsLink = dataSourceHDFSParam.getDsLink();
            String connectionProperty = dataSourceHDFSParam.getConnectionProperty();
            try {
                HdfsUtil.testConnection(hdfsUser, dsLink, connectionProperty);
                dataSourceEntity.setHdfsConfig(hdfsConfig);
                dataSourceEntity.setStatus(1);
            } catch (Exception e) {
                dataSourceEntity.setStatus(0);
                log.error("连接失败\n" + e.getMessage());
            }
        } else if (dsType.equalsIgnoreCase(DSType.OpenTSDB.name())) {
            boolean b = OpenTSDBUtil.testConnection(param.getDsLink());
            dataSourceEntity.setStatus(b ? 1 : 0);
//        } else if (dsType.equalsIgnoreCase(DSType.MaxCompute.name())) {
//            DataSourceMaxComputeParam dataSourceMaxComputeParam =
//                (DataSourceMaxComputeParam) param;
//            String endPoint = dataSourceMaxComputeParam.getEndPoint();
//            String accessId = dataSourceMaxComputeParam.getAccessId();
//            String accessKey = dataSourceMaxComputeParam.getAccessKey();
//            String project = dataSourceMaxComputeParam.getProject();
//            dataSourceEntity.setStatus(1);
//            MaxComputeConfig maxComputeConfig =
//                ConvertUtil.copyProperties(dataSourceMaxComputeParam, MaxComputeConfig.class);
//            dataSourceEntity.setMaxComputeConfig(maxComputeConfig);
//            try {
//                testMaxCompute(accessId, accessKey, endPoint, project);
//            } catch (Exception e) {
//                dataSourceEntity.setStatus(0);
//            }
        } else if (dsType.equalsIgnoreCase(DSType.Dgraph.name())) {

            try {
                DgraphUtil.testConnection(dsLink, null);
                dataSourceEntity.setStatus(1);
            } catch (Exception e) {
                dataSourceEntity.setStatus(0);
            }
        } else if (dsType.equalsIgnoreCase(DSType.IoTDB.name())) {
            DataSourceIotDBSaveParam dataSourceIotDBParam = (DataSourceIotDBSaveParam) param;
            RdbmsConfig iotdbConfig = rdbmsConfigMapper.paramToConfig(dataSourceIotDBParam);
            List<String> jdbcUrlList =
                Arrays.asList(iotdbConfig.getServer().split(",")).stream()
                    .filter(StringUtils::isNotBlank)
                    .map(s -> "jdbc:iotdb://" + s)
                    .collect(Collectors.toList());
            // 集群有问题
            // dsLink = "jdbc:iotdb://"+iotdbConfig.getServer();
            dsLink = String.join(",", jdbcUrlList);
            dataSourceEntity.setDsLink(dsLink);
            dataSourceEntity.setRdbmsConfig(iotdbConfig);
            try {
                IoTDBOptions iotdbOptions = ioTDBOptionsMapper.paramToOptions(dataSourceIotDBParam);
                IoTDBSessionManager iotdbManager = new IoTDBSessionManager(iotdbOptions);
                iotdbManager.testConnection();
            } catch (Exception e) {
                status = 0;
                log.error("连接iotdb异常，错误：{}", e);
            }
            dataSourceEntity.setStatus(status);
        } else if (dsType.equalsIgnoreCase(DSType.InfluxDB.name())) {
            DataSourceInfluxDBSaveParam influxDBParam = (DataSourceInfluxDBSaveParam) param;
            RdbmsConfig influxDbConfig = rdbmsConfigMapper.paramToConfig(influxDBParam);
            dataSourceEntity.setRdbmsConfig(influxDbConfig);

            // TODO: 暂时不支持InfluxDB Clustered集群版本
            // 为避免用户提供的server中不包含http协议，自动为host:ip格式的server开头补上http://
            if (!influxDbConfig.getServer().startsWith("http")) {
                dsLink = "http://" + influxDbConfig.getServer();
            } else {
                dsLink = influxDbConfig.getServer();
            }
            dataSourceEntity.setDsLink(dsLink);

            try {
                InfluxDbOptions influxDbOptions =
                    InfluxDbOptions.builder()
                        .url(dsLink)
                        .apiToken(influxDBParam.getApiToken())
                        .orgName(influxDBParam.getOrgName())
                        .dbName(influxDBParam.getDbName())
                        .build();
                InfluxDBManager influxDBManager = DbManagerFactory.getInfluxDbManager(influxDbOptions);
                influxDBManager.testConnection();
            } catch (SQLException e) {
                status = 0;
                log.error("连接InfluxDB异常，错误：{}", e);
            }
            dataSourceEntity.setStatus(status);
        }

        dataSourceEntity.setUpdateTime(new Date());
        dataSourceEntity.setLastUpdate(new Date());
        dataSourceEntity.setRowState(1);
        if (dataSourceEntity.getProjectId() == null) {
            dataSourceEntity.setProjectId(ServletUtils.getProjectId());
        }

        // exit
        if (flag) {
            updateById(dataSourceEntity);
//            LogContext.put(LogKey.ISUPDATE.getKey(), LogKey.ISUPDATE.getValue());
//            LogContext.put(LogKey.NAME.getKey(), param.getDsName());
        } else {
            this.save(dataSourceEntity);
            Long dataSourceId = dataSourceEntity.getId();
            param.setId(dataSourceId);
//            LogContext.put(LogKey.NAME.getKey(), param.getDsName());
            return dataSourceId;
        }
        return param.getId();
    }

    @Override
    public boolean dropDataSourceById(Long id, Long projectId) {
        // TODO: delete offline sync and dev records
//        List<OfflineSyncSimpleDomain> offlineSyncSimpleDomain =
//            offlineSyncService.getOfflineSyncSimpleDomain(id, projectId);
//
//        StringBuilder sb = new StringBuilder();
//        boolean flag = false;
//        if (org.apache.commons.collections4.CollectionUtils.isNotEmpty(offlineSyncSimpleDomain)) {
//            sb.append("无法删除该数据源在数据集成任务中使用:");
//            for (OfflineSyncSimpleDomain syncSimpleDomain : offlineSyncSimpleDomain) {
//                sb.append("\n").append(syncSimpleDomain.getTaskName());
//            }
//            flag = true;
//        }
//
//        List<DatadevOfflineSyncEntity> offlineDevSyncDomain =
//            datadevOfflineSyncService.listDatadevSyncTask(id);
//        if (org.apache.commons.collections4.CollectionUtils.isNotEmpty(offlineDevSyncDomain)) {
//            sb.append("无法删除该数据源在离线开发-数据集成任务中使用:");
//            for (DatadevOfflineSyncEntity syncSimpleDomain : offlineDevSyncDomain) {
//                sb.append("\n").append(syncSimpleDomain.getTaskName());
//            }
//            flag = true;
//        }
//
//        if (flag) {
//            throw new BusinessException(sb.toString());
//        }
        return this.removeById(id);
    }

    @Override
    public List<DataSourceDomain> getSystemDataSourceByProjectIds(List<Long> projectIds) {
        if (CollectionUtils.isEmpty(projectIds)) {
            return List.of();
        }
        List<DataSourceEntity> dataSourceEntities =
            this.list(
                new LambdaQueryWrapper<DataSourceEntity>()
                    .eq(DataSourceEntity::getIsSystem, 1)
                    .in(
                        CollectionUtils.isNotEmpty(projectIds),
                        DataSourceEntity::getProjectId,
                        projectIds));

        return dataSourceDomainMapper.map(dataSourceEntities);
    }

    @Override
    public DataSourceManagerDomain getManagerInfo(Long dsId) {
        DataSourceDetailDomain base = this.getDataSourceDetailById(dsId, false);
        if (Objects.isNull(base)) {
            return null;
        }
        String dsType = base.getDsType();
        String username = base.getUsername();
        String password = base.getPassword();
        String dsLink = base.getDsLink();
        String dbName = base.getDbName();
        Integer isSystem = base.getIsSystem();
        String schema = "";

        if (DBUtil.checkRDBMS(dsType) && !Objects.isNull(base.getRdbmsConfig())) {
            schema = base.getRdbmsConfig().getSchema();
        }

        String hiveSiteAddress =
            Optional.ofNullable(base.getHiveConfig())
                .map(DataSourceDetailDomain.HiveConfig::getHiveSiteAddress)
                .orElse("");
        Database database = Database.from(dsType);

        DataSourceManagerDomain manager =
            DataSourceManagerDomain.builder()
                .dbName(dbName)
                .dsType(dsType)
                .isSystem(isSystem)
                .schema(schema)
                .build();
        if (!Objects.isNull(database)
            && !dsType.equalsIgnoreCase(DSType.IoTDB.name())
            && !dsType.equalsIgnoreCase(DSType.InfluxDB.name())) { // 关系型
            // if (DSType.isRdbms(dsType)) { // 关系型
            DbOptions dbOptions = new DbOptions(database, username, password, dsLink, hiveSiteAddress);
            dbOptions.setSchema(schema);
            BaseDbManager dbManager = DbManagerFactory.getDbManager(database, dbOptions);
            manager.setDbManager(dbManager);
            return manager;
        }

        if (dsType.equalsIgnoreCase(DSType.Mongodb.name())) { // mongo
            MongoOptions mongoOptions = new MongoOptions();
            mongoOptions.setAuthDbName(base.getMongoConfig().getAuthDbName());
            mongoOptions.setUsername(username);
            mongoOptions.setPassword(password);
            mongoOptions.setDbName(dbName);
            mongoOptions.setAddressList(Collections.singletonList(dsLink));
            MongoManager mongoManager = DbManagerFactory.getMongoManager(mongoOptions);
            manager.setMongoManager(mongoManager);
            return manager;
        }

        if (dsType.equalsIgnoreCase(DSType.HBase.name())) { // hbase
            HBaseManager hBaseManager = DbManagerFactory.getHBaseManager(dsLink);
            manager.setHBaseManager(hBaseManager);
            return manager;
        }

        if (dsType.equalsIgnoreCase(DSType.IoTDB.name())) { // iotdb
            IoTDBOptions iotdbOptions = new IoTDBOptions();
            iotdbOptions.setServer(base.getRdbmsConfig().getServer());
            iotdbOptions.setUsername(base.getUsername());
            iotdbOptions.setPassword(base.getPassword());
            IoTDBSessionManager iotdbManager = new IoTDBSessionManager(iotdbOptions);
            manager.setIoTDBSessionManager(iotdbManager);
            return manager;
        }

        if (dsType.equalsIgnoreCase(DSType.InfluxDB.name())) { // influxdb
            InfluxDbOptions influxDbOptions = new InfluxDbOptions();
            influxDbOptions.setUrl(dsLink);
            influxDbOptions.setApiToken(password);
            influxDbOptions.setOrgName(base.getRdbmsConfig().getSchema());
            influxDbOptions.setDbName(base.getDbName());
            InfluxDBManager influxDbManager = DbManagerFactory.getInfluxDbManager(influxDbOptions);
            manager.setInfluxDBManager(influxDbManager);
            return manager;
        }

        return manager;
    }

    @Override
    public Boolean tableExist(Long dsId, String tableName) {
        DataSourceManagerDomain managerInfo = this.getManagerInfo(dsId);
        BaseDbManager dbManager = managerInfo.getDbManager();
        if (Objects.isNull(dbManager)) {
            throw new BusinessException(BizCodes.DS_NOT_EXISTS_ERROR, HttpStatus.BAD_REQUEST);
        }
        TableInfoDomain tableInfo = null;
        try {
            tableInfo = dbManager.getTableInfoByTableName(managerInfo.getDbName(), tableName);
        } catch (SQLException throwables) {
            log.error("获取表信息失败：", throwables);
        }
        return Objects.isNull(tableInfo) ? Boolean.FALSE : Boolean.TRUE;
    }

    private boolean doesDsNameExist(DataSourceSaveParam param) {
        DataSourceEntity dataSourceEntity = dataSourceEntityMapper.saveParamToEntity(param);
        String dsType = dataSourceEntity.getDsType();
        Long projectId = ServletUtils.getProjectId();
        String dsName = dataSourceEntity.getDsName();

        LambdaQueryWrapper<DataSourceEntity> lambdaQueryWrapper =
            new LambdaQueryWrapper<DataSourceEntity>()
                .eq(DataSourceEntity::getDsType, dsType)
                .eq(Objects.nonNull(projectId), DataSourceEntity::getProjectId, projectId)
                .eq(DataSourceEntity::getDsName, dsName)
                .ne(
                    Objects.nonNull(param.getId()),
                    DataSourceEntity::getId,
                    param.getId());

        List<DataSourceEntity> multi = this.list(lambdaQueryWrapper);

        if (param.getId() == null) {
            // 新增数据源
            return CollectionUtils.isNotEmpty(multi);
        }

        // 更新数据源，除了本身没有其它相同的即可
        return !multi.isEmpty();
    }

    @Override
    public Boolean testDataSourceConnection(DataSourceConnectParam dataSourceConnectParam) {

        // 测试rdbms数据源和hive
        String dsType = dataSourceConnectParam.getDsType();
        String username = dataSourceConnectParam.getUsername();
        String password = dataSourceConnectParam.getPassword();
        String dsLink = dataSourceConnectParam.getDsLink();

        if (StrUtil.containsBlank(dsLink)) { // 数据库连接link包含空格时，通过jdbc driver可以连接成功，但在使用连接池的情况会连接失败
            throw new BusinessException(BizCodes.INVALID_DS_LINK_ERROR, HttpStatus.BAD_REQUEST);
        }
        if (DSType.isRdbms(dsType)) {
            try {
                BaseDbManager dbManager = getDbManager(dsType, username, password, dsLink);
                dbManager.testConnection();
            } catch (SQLException e) {
                log.error(e.getMessage(), e);
                throw new BusinessException(BizCodes.DS_CONNECTION_FAILED, HttpStatus.BAD_REQUEST);
            }

        } else if (dsType.equalsIgnoreCase(DSType.Hive.name())) {
            DataSourceHiveConnectParam hiveConnectParam = (DataSourceHiveConnectParam) dataSourceConnectParam;
            try {
                BaseDbManager dbManager =
                    getDbManager(dsType, username, password, dsLink, hiveConnectParam.getHiveSiteAddress());
                dbManager.testConnection();
            } catch (Exception e) {
                log.error(e.getMessage());
                throw new BusinessException(BizCodes.DS_CONNECTION_FAILED, HttpStatus.BAD_REQUEST);
            }
        } else if (dsType.equalsIgnoreCase(DSType.HBase.name())) {
            try {
                HBaseUtil.testConnection(dsLink);
            } catch (Exception e) {
                log.error(e.getMessage());
                throw new BusinessException(BizCodes.DS_CONNECTION_FAILED, HttpStatus.BAD_REQUEST);
            }
        } else if (dsType.equalsIgnoreCase(DSType.Mongodb.name())) {
            DataSourceMongoConnectParam mongodbParam = (DataSourceMongoConnectParam) dataSourceConnectParam;
            MongoOptions mongoOptions = new MongoOptions();
            mongoOptions.setAddressList(List.of(dsLink));
            mongoOptions.setAuthDbName(mongodbParam.getAuthDbName());
            mongoOptions.setDbName(mongodbParam.getDbName());
            mongoOptions.setUsername(username);
            mongoOptions.setPassword(password);
            MongoManager mongoManager = new MongoManager(mongoOptions);
            try {
                mongoManager.testConnection();
            } catch (Exception e) {
                log.error(e.getMessage());
                throw new BusinessException(BizCodes.DS_CONNECTION_FAILED, HttpStatus.BAD_REQUEST);
            }
        } else if (dsType.equalsIgnoreCase(DSType.MQTT.name())) {
            try {
                MQTTUtil.testConnection(dsLink, null);
            } catch (Exception e) {
                log.error(e.getMessage());
                throw new BusinessException(BizCodes.DS_CONNECTION_FAILED, HttpStatus.BAD_REQUEST);
            }
        } else if (dsType.equalsIgnoreCase(DSType.Kafka.name())) {
            try {
                KafkaUtil.testConnection(dsLink, null);
            } catch (Exception e) {
                log.error(e.getMessage());
                throw new BusinessException(BizCodes.DS_CONNECTION_FAILED, HttpStatus.BAD_REQUEST);
            }
        } else if (dsType.equalsIgnoreCase(DSType.HDFS.name())) {
            DataSourceHDFSConnectParam dataSourceHDFSConnectParam = (DataSourceHDFSConnectParam) dataSourceConnectParam;
            String connectionProperty = dataSourceHDFSConnectParam.getConnectionProperty();
            String hdfsUser = dataSourceHDFSConnectParam.getHdfsUser();
            try {
                HdfsUtil.testConnection(hdfsUser, dsLink, connectionProperty);
            } catch (Exception e) {
                log.error("连接失败\n{}", e.getMessage());
                throw new BusinessException(BizCodes.DS_CONNECTION_FAILED, HttpStatus.BAD_REQUEST);
            }
        } else if (dsType.equalsIgnoreCase(DSType.OpenTSDB.name())) {
            boolean b = OpenTSDBUtil.testConnection(dataSourceConnectParam.getDsLink());
            if (!b) {
                throw new BusinessException(BizCodes.DS_CONNECTION_FAILED, HttpStatus.BAD_REQUEST);
            }
        } else if (dsType.equalsIgnoreCase(DSType.Dgraph.name())) {
            try {
                DgraphUtil.testConnection(dsLink, 3000);
            } catch (Exception e) {
                log.error(e.getMessage());
                throw new BusinessException(BizCodes.DS_CONNECTION_FAILED, HttpStatus.BAD_REQUEST);
            }
        } else if (dsType.equalsIgnoreCase(DSType.IoTDB.name())) {
            try {
                var dataSourceIoTDBConnectParam = (DataSourceIoTDBConnectParam) dataSourceConnectParam;
                IoTDBOptions iotdbOptions = ioTDBOptionsMapper.paramToOptions(dataSourceIoTDBConnectParam);
                IoTDBSessionManager iotdbManager = new IoTDBSessionManager(iotdbOptions);
                iotdbManager.testConnection();
            } catch (Exception e) {
                log.error("连接iotdb异常，错误：{}", e.getMessage());
                throw new BusinessException(BizCodes.DS_CONNECTION_FAILED, HttpStatus.BAD_REQUEST);
            }
        } else if (dsType.equalsIgnoreCase(DSType.InfluxDB.name())) {
            try {
                var dataSourceInfluxDBConnectParam = (DataSourceInfluxDBConnectParam) dataSourceConnectParam;
                if (!dataSourceInfluxDBConnectParam.getServer().startsWith("http")) {
                    dsLink = "http://" + dataSourceInfluxDBConnectParam.getServer();
                } else {
                    dsLink = dataSourceInfluxDBConnectParam.getServer();
                }
                InfluxDbOptions influxDbOptions =
                    InfluxDbOptions.builder()
                        .url(dsLink)
                        .apiToken(dataSourceInfluxDBConnectParam.getApiToken())
                        .orgName(dataSourceInfluxDBConnectParam.getOrgName())
                        .dbName(dataSourceInfluxDBConnectParam.getDbName())
                        .build();
                InfluxDBManager influxDbManager = DbManagerFactory.getInfluxDbManager(influxDbOptions);
                influxDbManager.testConnection();
            } catch (SQLException e) {
                log.error("连接InfluxDB异常，错误：{}", e.getMessage());
                throw new BusinessException(BizCodes.DS_CONNECTION_FAILED, HttpStatus.BAD_REQUEST);
            }
        }

        if (Objects.nonNull(dataSourceConnectParam.getId())) {
            DataSourceEntity dataSourceEntity =
                this.baseMapper.selectById(dataSourceConnectParam.getId());
            dataSourceEntity.setStatus(1);
            dataSourceEntity.setLastUpdate(new Date());

            this.updateById(dataSourceEntity);
        }

        return true;
    }

    @Override
    public List<DataSourceEntity> getDataSourceActive() {
        return this.list(
            new LambdaQueryWrapper<DataSourceEntity>().eq(DataSourceEntity::getRowState, 1));
    }

    @Override
    public void updateStatusByIds(int status, List<Long> successDS) {
        this.update(
            new LambdaUpdateWrapper<DataSourceEntity>()
                .in(DataSourceEntity::getId, successDS)
                .set(DataSourceEntity::getStatus, status));
    }

    @Override
    public String getLastUpdateTime() {
        Long projectId = ServletUtils.getProjectId();
        ProjectEntity projectById = projectService.getProjectById(projectId);
        String s = redisTemplate.opsForValue().get(RedisKey.DATASOURCE_LAST_UPDATE);
        if (StringUtils.isNotBlank(s) && Objects.nonNull(projectById.getUpdateTime())) {
            Date date = DataUtil.parse(s);
            if (date.before(projectById.getUpdateTime())) {
                return null;
            }
        }
        return s;
    }

    @Override
    public DataSourceDomain getSystemDataSourceByProjectId(Long projectId) {
        LambdaQueryWrapper<DataSourceEntity> lambdaQueryWrapper =
            new LambdaQueryWrapper<DataSourceEntity>()
                .eq(DataSourceEntity::getIsSystem, 1)
                .eq(DataSourceEntity::getProjectId, projectId);

        DataSourceEntity dataSourceEntity = this.baseMapper.selectOne(lambdaQueryWrapper);
        return dataSourceDomainMapper.map(dataSourceEntity);
    }

    @Override
    public DataSourceEntity getSystemDataSourceDetail(Long projectId) {
        LambdaQueryWrapper<DataSourceEntity> lambdaQueryWrapper =
            new LambdaQueryWrapper<DataSourceEntity>()
                .eq(DataSourceEntity::getIsSystem, 1)
                .eq(DataSourceEntity::getProjectId, projectId);

        return this.baseMapper.selectOne(lambdaQueryWrapper);
    }

    @Override
    public DataSourceDetailDomain getDataSourceDetailByProjectId(Long projectId, boolean encrypt) {
        DataSourceDomain systemDataSourceByProjectId = this.getSystemDataSourceByProjectId(projectId);
        if (null == systemDataSourceByProjectId) {
            return null;
        }
        return getDataSourceDetailById(systemDataSourceByProjectId.getId(), encrypt);
    }

    @Override
    public List<DataSourceDomain> getAllByProject(Long projectId) {
        List<DataSourceEntity> all =
            list(
                new LambdaQueryWrapper<DataSourceEntity>()
                    .eq(DataSourceEntity::getProjectId, projectId)
                    .eq(DataSourceEntity::getRowState, 1));

        return dataSourceDomainMapper.map(all);
    }

    @Override
    public List<DataSourceMultiDomain> getDSByParams(DataSourceQueryDomain queryDomain) {

        List<DataSourceEntity> dsByParams =
            this.list(
                new LambdaQueryWrapper<DataSourceEntity>()
                    .eq(
                        Objects.nonNull(queryDomain.getProjectId()),
                        DataSourceEntity::getProjectId,
                        queryDomain.getProjectId())
                    .in(
                        org.apache.commons.collections4.CollectionUtils.isNotEmpty(queryDomain.getDsTypes()),
                        DataSourceEntity::getDsType,
                        queryDomain.getDsTypes()));

        ArrayList<DataSourceMultiDomain> dataSourceDomains = Lists.newArrayList();
        dsByParams.forEach(
            dsByParam -> {
                DataSourceMultiDomain dataSourceDomain = dataSourceMultiDomainMapper.map(dsByParam);
                String dsType = dataSourceDomain.getDsType();
                if (DSType.isRdbms(dsType)) {
                    dataSourceDomain.setPassword(dsByParam.getRdbmsConfig().getPassword());
                    dataSourceDomain.setUsername(dsByParam.getRdbmsConfig().getUsername());
                    dataSourceDomain.setDbName(dsByParam.getRdbmsConfig().getDbName());
                } else if (DSType.Hive.name().equals(dsType)) {
                    dataSourceDomain.setPassword(dsByParam.getHiveConfig().getPassword());
                    dataSourceDomain.setUsername(dsByParam.getHiveConfig().getUsername());
                    dataSourceDomain.setDbName(dsByParam.getHiveConfig().getDbName());
                } else if (DSType.Mongodb.name().equals(dsType)) {
                    dataSourceDomain.setPassword(dsByParam.getMongoConfig().getPassword());
                    dataSourceDomain.setUsername(dsByParam.getMongoConfig().getUsername());
                } else if (DSType.MaxCompute.name().equals(dsType)) {
                    dataSourceDomain.setPassword(dsByParam.getMaxComputeConfig().getAccessId());
                    dataSourceDomain.setUsername(dsByParam.getMaxComputeConfig().getAccessKey());
                }
                dataSourceDomains.add(dataSourceDomain);
            });
        return dataSourceDomains;
    }

    /**
     * 获取数据源指定表的字段信息
     *
     * @param dsId
     * @param tableName
     * @return
     */
    @Override
    public List<FieldInfoDomain> getFieldList(Long dsId, String tableName) {
        DataSourceEntity dataSourceEntity = this.baseMapper.selectById(dsId);
        String dsType = dataSourceEntity.getDsType();
        String dsLink = dataSourceEntity.getDsLink();
        List<FieldInfoDomain> fieldInfoDomainList;

        DataSourceManagerDomain managerInfo = getManagerInfo(dsId);
        if (null == managerInfo) {
            throw new BusinessException("获取表字段失败: " + dsId + "不存在");
        }

        String dbName = managerInfo.getDbName();
        final BaseDbManager dbManager = managerInfo.getDbManager();
        final MongoManager mongoManager = managerInfo.getMongoManager();
        final IoTDBSessionManager ioTDBSessionManager = managerInfo.getIoTDBSessionManager();
        final InfluxDBManager influxDBManager = managerInfo.getInfluxDBManager();

        if (!Objects.isNull(dbManager)) {
            if (DSType.isRdbms(dsType)) {
                RdbmsConfig rdbmsConfig = dataSourceEntity.getRdbmsConfig();
                if (null == rdbmsConfig) {
                    throw new BusinessException("check datasource config");
                }
                dbName = rdbmsConfig.getDbName();
            } else if (DSType.Hive.equals(DSType.valueOf(dsType))) {
                dbName = managerInfo.getDbName();
            }
            boolean flag = dbManager.hasSchema();
            try {
                if (flag && tableName.contains(".")) {
                    String[] split = tableName.split("\\.", 2);
                    fieldInfoDomainList = dbManager.getFieldList(dbName, split[0], split[1]);
                } else if (DSType.DM.equals(DSType.valueOf(dsType)) && tableName.contains(".")) {
                    String[] split = tableName.split("\\.", 2);
                    fieldInfoDomainList = dbManager.getFieldList(dbName, split[1]);
                } else {
                    fieldInfoDomainList = dbManager.getFieldList(dbName, tableName);
                }
                return fieldInfoDomainList;
            } catch (SQLException e) {
                log.error(e.getMessage());
                throw new BusinessException("获取表字段失败: " + e.getMessage());
            } finally {
                dbManager.close();
            }
        }
        if (!Objects.isNull(mongoManager)) {
            fieldInfoDomainList = mongoManager.getFieldList(dbName, tableName);
            return fieldInfoDomainList;
        }
        if (!Objects.isNull(ioTDBSessionManager)) {
            try {
                fieldInfoDomainList = ioTDBSessionManager.getFiledList(String.join(".", tableName, "**"));
            } catch (Exception e) {
                log.error("获取iotdb字段信息失败，{}", e);
                fieldInfoDomainList = null;
            }
            return fieldInfoDomainList;
        }
        if (!Objects.isNull(influxDBManager)) {
            try {
                fieldInfoDomainList = influxDBManager.getFieldList(dbName, tableName);
            } catch (Exception e) {
                log.error("获取influxdb字段信息失败，{}", e);
                fieldInfoDomainList = null;
            }
            return fieldInfoDomainList;
        }
        throw new BusinessException("检查数据源状态");
    }

    @Override
    public List<String> getDataSourceSchema(Long dsId, String dsType) {
        if (!DSType.hasSchema(dsType)) {
            return Lists.newArrayList();
        }
        DataSourceManagerDomain managerInfo = this.getManagerInfo(dsId);
        BaseDbManager dbManager = managerInfo.getDbManager();
        try {
            if (null != dbManager) {
                return dbManager.getSchemaList();
            }
        } catch (SQLException e) {
            try {
                dbManager.testConnection();
            } catch (SQLException ex) {
                log.error(ex.getMessage());
                throw new BusinessException("获取schema失败,请检测网络和数据库服务");
            }
            return Lists.newArrayList();
        }
        return Lists.newArrayList();
    }

    @Override
    public String getCreateTableSql(
        Long dsId, String tableName, Long targetDsId, String schema, int code) {

        DataSourceManagerDomain targetManagerInfo = this.getManagerInfo(targetDsId);

        Integer isSystem = targetManagerInfo.getIsSystem();
        if (!Objects.isNull(isSystem) && 0 == isSystem) {
            throw new BusinessException("一键建表仅支持底层存储数据源，请检查后重试！");
        }

        DataSourceManagerDomain sourceManagerInfo = this.getManagerInfo(dsId);
        //    if (sourceManagerInfo.getDsType().equals(DSType.SQLServer.name()) &&
        // tableName.contains(".")) {
        //      String[] names = tableName.split("\\.");
        //      tableName = names[1];
        //    }
        if (Objects.isNull(sourceManagerInfo)) {
            throw new BusinessException("获取数据源信息失败，请检查后重试！");
        }

        String dsType = sourceManagerInfo.getDsType();
        if (dsType.equalsIgnoreCase(DSType.HBase.name())) {
            throw new BusinessException("hbase来源不支持一键建表！");
        }

        BaseDbManager dbManager = sourceManagerInfo.getDbManager();
        MongoManager mongoManager = sourceManagerInfo.getMongoManager();
        String dbName = sourceManagerInfo.getDbName();
        String targetDB = targetManagerInfo.getDbName();
        if (sourceManagerInfo.getDsType().equals(DSType.Sap.name())) {
            dbName = sourceManagerInfo.getSchema();
        }
        TableInfoDomain tableInfoDomain = null;
        List<FieldInfoDomain> fieldList = null;
        try {
            if (!Objects.isNull(dbManager)) {
                boolean flag = dbManager.hasSchema();
                String schemaSource = null;
                if (flag && tableName.contains(".")) {
                    String[] split = tableName.split("\\.", 2);
                    tableName = split[1];
                    schemaSource = split[0];
                }
                if (dsType.equalsIgnoreCase(DSType.DM.name())) {
                    if (tableName.contains(".")) {
                        String[] splitDM = tableName.split("\\.", 2);
                        tableName = splitDM[1];
                        dbName = splitDM[0];
                    }
                }
                if (StringUtils.isNotBlank(schemaSource)) {
                    tableInfoDomain = dbManager.getTableInfoByTableName(dbName, schemaSource, tableName);
                    fieldList = dbManager.getFieldList(dbName, schemaSource, tableName);
                } else {
                    tableInfoDomain = dbManager.getTableInfoByTableName(dbName, tableName);
                    fieldList = dbManager.getFieldList(dbName, tableName);
                }
            } else if (!Objects.isNull(mongoManager)) {
                tableInfoDomain = new TableInfoDomain();
                tableInfoDomain.setTableName(tableName);
                fieldList = mongoManager.getFieldList(dbName, tableName);
            }
        } catch (SQLException throwables) {
            log.error("离线同步，获取来源数据表字段异常：", throwables);
            throw new BusinessException(
                "离线同步，获取来源数据表信息异常：" + throwables.getMessage());
        }
        if (Objects.isNull(tableInfoDomain) || org.apache.commons.collections4.CollectionUtils.isEmpty(fieldList)) {
            throw new BusinessException("来源数据表不存在或表字段为空，请检查后重试！");
        }

        BaseDbManager targetDbManager = targetManagerInfo.getDbManager();

        CreateTableParam param = new CreateTableParam();
        if (StringUtils.isNotBlank(schema)) {
            param.setSchema(schema);
        } else if (targetDbManager.hasSchema()) {
            param.setSchema(targetManagerInfo.getSchema());
        }
        param.setDbName(targetDB);
        String targetName = tableInfoDomain.getTableName();
        targetName = StringUtils.replace(targetName, ".", "_");
        targetName = "ods_" + targetName;
        if (CreateTableCaseEnum.UPPERCASE.getCode() == code) {
            targetName = targetName.toUpperCase(Locale.ROOT);
        } else if (CreateTableCaseEnum.LOWERCASE.getCode() == code) {
            targetName = targetName.toLowerCase(Locale.ROOT);
        }
        param.setTableName(targetName);
        List<FieldInfoDomain> r1 =
            fieldList.stream()
                .map(
                    e -> {
                        if (CreateTableCaseEnum.LOWERCASE.getCode() == code) {
                            e.setFieldName(e.getFieldName().toLowerCase(Locale.ROOT));
                        } else if (CreateTableCaseEnum.UPPERCASE.getCode() == code) {
                            e.setFieldName(e.getFieldName().toUpperCase(Locale.ROOT));
                        }
                        e.setFieldType(targetDbManager.toDbType(e.getDataType(), e.getFieldType(), e.getColumnSize(), e.getDecimalDigits()));
                        return e;
                    })
                .collect(Collectors.toList());
        List<CreateTableParam.FieldInfo> fieldInfoList =
            r1.stream().map(this::toFieldInfo).collect(Collectors.toList());
        param.setFieldInfoList(fieldInfoList);
        String sql = "";
        try {
            DDL ddl = targetDbManager.getDDLBuilder().create(param);
            sql = ddl.toString();
        } catch (SQLException throwables) {
            log.error("建表异常", throwables);
        }
        if (StringUtils.isNotEmpty(sql)) {
            return StringUtils.trimToEmpty(sql);
        }
        return null;
    }

    @Override
    public PreviewDataDomain previewData(Long dsId, String tableName) {
        DataSourceManagerDomain managerInfo = this.getManagerInfo(dsId);
        if (Objects.isNull(managerInfo)) {
            throw new BusinessException("获取数据源信息失败，请检查后重试！");
        }
//        boolean secretLevel =
//            tblPropService.checkSecretLevel(
//                tableName, ServletUtils.getProjectId(), UserContext.getUserId());
//
//        if (!secretLevel) {
//            userBehaviorLogService.addFailedLog(
//                Business.数据资产全域数据, Operation.PRE_VIEW, tableName, LogResult.FAILED);
//            throw new BusinessException("密级不符合");
//        }
        BaseDbManager dbManager = managerInfo.getDbManager();
        MongoManager mongoManager = managerInfo.getMongoManager();
        HBaseManager hBaseManager = managerInfo.getHBaseManager();
        IoTDBSessionManager ioTDBSessionManager = managerInfo.getIoTDBSessionManager();
        InfluxDBManager influxDBManager = managerInfo.getInfluxDBManager();
        PreviewDataDomain previewDataDomain = null;
        try {
            if (!Objects.isNull(dbManager)) {
                boolean flag = dbManager.hasSchema();
                if (flag && tableName.contains(".")) {
                    String[] split = tableName.split("\\.", 2);
                    String schema = split[0];
                    tableName = split[1];
                    previewDataDomain = dbManager.previewData(schema, tableName);
                } else {
                    previewDataDomain = dbManager.previewData(tableName);
                }
            } else if (!Objects.isNull(mongoManager)) {
                previewDataDomain = mongoManager.previewData(tableName);
            } else if (!Objects.isNull(hBaseManager)) {
                previewDataDomain = hBaseManager.previewData(tableName, true);
            } else if (!Objects.isNull(ioTDBSessionManager)) {
                previewDataDomain = ioTDBSessionManager.previewData(tableName);
            } else if (!Objects.isNull(influxDBManager)) {
                previewDataDomain = influxDBManager.previewData(tableName);
            }
        } catch (Exception throwables) {
            log.error("离线同步，数据预览异常：", throwables);
        }
        return previewDataDomain;
    }

    /**
     * @param dsId
     * @param createTableSql
     * @param dataSecretId   选择的数据密级
     * @return
     */
    @Override
    public DataSourceCreateTableDomain executeCreateTableSql(
        Long dsId, String createTableSql, Long dataSecretId) {
        DataSourceManagerDomain managerInfo = this.getManagerInfo(dsId);
        Integer isSystem = managerInfo.getIsSystem();
        if (!Objects.isNull(isSystem) && 0 == isSystem) {
            throw new BusinessException("一键建表仅支持底层存储数据源，请检查后重试！");
        }

        BaseDbManager dbManager = managerInfo.getDbManager();
        String dsType = managerInfo.getDsType();

        if (DSType.Sap.name().equalsIgnoreCase(dsType)) {
            dsType = DbType.sapdb.name();
        }
        if (DSType.TiDB.name().equals(dsType)) {
            dsType = DbType.mysql.name();
        }
        if (DSType.KINGBASE8.name().equals(dsType)) {
            dsType = DbType.kingbase.name();
        }
        if (DSType.DM.name().equals(dsType)) {
            dsType = DbType.dm.name();
        }
        DbType dbType = EnumUtil.fromString(DbType.class, dsType.toLowerCase());

        // TODO SAP HANA暂不支持语法解析，直接跳过解析这一步
        // hive转义会报错, 跳过正则替换步骤
        if (dbType != DbType.sapdb && dbType != DbType.hive) {
            Matcher matcher = pattern.matcher(createTableSql);
            String createSqlCopy = createTableSql;
            while (matcher.find()) {
                String originalContent = matcher.group(1);
                // 替换描述里面的单引号
                String revisedContent = originalContent.replaceAll("'", "\\\\'");
                createSqlCopy = createSqlCopy.replace(originalContent, revisedContent);
            }
            createTableSql = createSqlCopy;
            List<String> scripts = SqlParserUtil.split(createTableSql, dbType);
            if (org.apache.commons.collections4.CollectionUtils.isEmpty(scripts)) {
                throw new BusinessException("请输入一条完整SQL语句!");
            }
        }

        if (dbType == DbType.hive) {
            List<String> scripts = SqlParserUtil.split(createTableSql, dbType);
            if (org.apache.commons.collections4.CollectionUtils.isEmpty(scripts)) {
                throw new BusinessException("请输入一条完整SQL语句!");
            }
        }

        // 移除Hive SQL影响的;
        if (StringUtils.trim(createTableSql).endsWith(";")) {
            createTableSql = createTableSql.substring(0, createTableSql.lastIndexOf(";"));
        }
        try (Connection connection = dbManager.getConnection();
             Statement stmt = connection.createStatement()) {
            stmt.execute(createTableSql);
        } catch (Exception e) {
            log.error("执行一键建表语句异常：", e);
            throw new BusinessException("执行一键建表语句异常：" + e.getMessage());
        }

        SQLStatementParser sqlStatementParser =
            SQLParserUtils.createSQLStatementParser(createTableSql, dbType);
        SQLCreateTableStatement sqlCreateTableStatement = sqlStatementParser.parseCreateTable();

        String tableName = sqlCreateTableStatement.getTableSource().getExpr().toString();
        tableName =
            tableName.replaceAll("\"", "").replaceAll("'", "").replaceAll("\"", "").replaceAll("`", "");

        DataSourceCreateTableDomain dataSourceCreateTableDomain = new DataSourceCreateTableDomain();
        dataSourceCreateTableDomain.setTableName(tableName);
        dataSourceCreateTableDomain.setOnlyTableName(tableName);
        dataSourceCreateTableDomain.setSchema(managerInfo.getSchema());
        if (DSType.hasSchema(managerInfo.getDsType()) && tableName.contains(".")) {
            String[] split = tableName.split("\\.", 2);
            dataSourceCreateTableDomain.setSchema(split[0]);
            dataSourceCreateTableDomain.setOnlyTableName(split[1]);
        }
        if (StringUtils.equalsIgnoreCase(DSType.Hive.name(), managerInfo.getDsType())
            || StringUtils.equalsIgnoreCase(DSType.TiDB.name(), managerInfo.getDsType())) {
            dataSourceCreateTableDomain.setOnlyTableName(tableName);
        }
        return dataSourceCreateTableDomain;
    }

    @Override
    public DataSourceCreateTableDomain executeCreateTableSql(Long dsId, String createTableSql) {
        DataSourceManagerDomain managerInfo = this.getManagerInfo(dsId);
        Integer isSystem = managerInfo.getIsSystem();
        if (!Objects.isNull(isSystem) && 0 == isSystem) {
            throw new BusinessException("一键建表仅支持底层存储数据源，请检查后重试！");
        }

        BaseDbManager dbManager = managerInfo.getDbManager();
        String dsType = managerInfo.getDsType();

        if (DSType.Sap.name().equalsIgnoreCase(dsType)) {
            dsType = DbType.sapdb.name();
        } else if (DSType.TiDB.name().equals(dsType)) {
            dsType = DbType.mysql.name();
        } else if (DSType.KINGBASE8.name().equalsIgnoreCase(dsType)) {
            dsType = DbType.kingbase.name();
        } else if (DSType.DM.name().equalsIgnoreCase(dsType)) {
            dsType = DbType.dm.name();
        }

        DbType dbType = EnumUtil.fromString(DbType.class, dsType.toLowerCase());

        // TODO SAP HANA暂不支持语法解析，直接跳过解析这一步
        // hive转义会报错, 跳过正则替换步骤
        if (dbType != DbType.sapdb && dbType != DbType.hive) {
            Matcher matcher = pattern.matcher(createTableSql);
            String createSqlCopy = createTableSql;
            while (matcher.find()) {
                String originalContent = matcher.group(1);
                // 替换描述里面的单引号
                String revisedContent = originalContent.replaceAll("'", "\\\\'");
                createSqlCopy = createSqlCopy.replace(originalContent, revisedContent);
            }
            createTableSql = createSqlCopy;
            List<String> scripts = SqlParserUtil.split(createTableSql, dbType);
            if (org.apache.commons.collections4.CollectionUtils.isEmpty(scripts)) {
                throw new BusinessException("请输入一条完整SQL语句!");
            }
        }

        if (dbType == DbType.hive) {
            List<String> scripts = SqlParserUtil.split(createTableSql, dbType);
            if (org.apache.commons.collections4.CollectionUtils.isEmpty(scripts)) {
                throw new BusinessException("请输入一条完整SQL语句!");
            }
            // 移除Hive SQL影响的;
            if (StringUtils.trim(createTableSql).endsWith(";")) {
                createTableSql = createTableSql.substring(0, createTableSql.lastIndexOf(";"));
            }
        }
        List<String> stringList = SqlParserUtil.split(createTableSql, dbType);

        try (Connection connection = dbManager.getConnection();
             Statement stmt = connection.createStatement()) {
            for (String sql : stringList) {
                stmt.execute(sql);
            }
        } catch (Exception e) {
            log.error("执行一键建表语句异常：", e);
            throw new BusinessException("执行一键建表语句异常：" + e.getMessage());
        }

        SQLStatementParser sqlStatementParser =
            SQLParserUtils.createSQLStatementParser(createTableSql, dbType);
        SQLCreateTableStatement sqlCreateTableStatement = sqlStatementParser.parseCreateTable();

        String tableName = sqlCreateTableStatement.getTableSource().getExpr().toString();
        tableName =
            tableName.replaceAll("\"", "").replaceAll("'", "").replaceAll("\"", "").replaceAll("`", "");

        DataSourceCreateTableDomain dataSourceCreateTableDomain = new DataSourceCreateTableDomain();
        dataSourceCreateTableDomain.setTableName(tableName);
        dataSourceCreateTableDomain.setOnlyTableName(tableName);
        dataSourceCreateTableDomain.setSchema(managerInfo.getSchema());
        if (DSType.hasSchema(managerInfo.getDsType()) && tableName.contains(".")) {
            String[] split = tableName.split("\\.", 2);
            dataSourceCreateTableDomain.setSchema(split[0]);
            dataSourceCreateTableDomain.setOnlyTableName(split[1]);
        }
        if (StringUtils.equalsIgnoreCase(DSType.Hive.name(), managerInfo.getDsType())
            || StringUtils.equalsIgnoreCase(DSType.TiDB.name(), managerInfo.getDsType())) {
            dataSourceCreateTableDomain.setOnlyTableName(tableName);
        }
        return dataSourceCreateTableDomain;
    }

    @Override
    public List<DataSourceDomain> getDataSourceDomainByIds(Set<Long> dsIds) {
        List<DataSourceEntity> multi = this.listByIds(dsIds);
        return dataSourceDomainMapper.map(multi);
    }

    @Override
    public List<DataSourceDomain> getListByDsTypes(Long projectId, List<String> dsTypes) {
        List<DataSourceEntity> list =
            this.list(
                new LambdaUpdateWrapper<DataSourceEntity>()
                    .eq(DataSourceEntity::getProjectId, projectId)
                    .in(DataSourceEntity::getDsType, dsTypes));

        return ConvertUtil.copyProperties(list, DataSourceDomain.class);
    }

    @Override
    public Long checkSpKafka(Long projectId) {

        List<DataSourceEntity> multi =
            this.list(
                new LambdaUpdateWrapper<DataSourceEntity>()
                    .eq(DataSourceEntity::getProjectId, projectId)
                    .eq(DataSourceEntity::getDsName, "Kafka_for算盘组件"));

        if (CollectionUtils.isNotEmpty(multi)) {
            return multi.get(0).getId();
        }
        return null;
    }

    @Override
    public List<FieldInfoDomain> getFieldListProjectId(String tableName, Long projectId) {
        DataSourceEntity dataSource = this.getSystemDataSourceDetail(projectId);
        if (Objects.isNull(dataSource)) {
            throw new BusinessException("该数据源系统数据源异常");
        }
        return getFieldList(dataSource.getId(), tableName);
    }

    @Override
    public List<TableInfoDomain> getTableAndViewList(Long dsId) {
        DataSourceEntity dataSourceEntity = this.baseMapper.selectById(dsId);
        if (null == dataSourceEntity) {
            throw new BusinessException("获取表列表失败: " + dsId + "不存在");
        }
        String dsType = dataSourceEntity.getDsType();
        String dsLink = dataSourceEntity.getDsLink();
        List<TableInfoDomain> tableList = Lists.newArrayList();
        if (DSType.isRdbms(dsType)) {
            RdbmsConfig rdbmsConfig = dataSourceEntity.getRdbmsConfig();
            String rdbmsPassword = rdbmsConfig.getPassword();
            String rdbmsUsername = rdbmsConfig.getUsername();
            String db = rdbmsConfig.getDbName();
            BaseDbManager dbManager = getDbManager(dsType, rdbmsUsername, rdbmsPassword, dsLink);
            try {
                if (dbManager.hasSchema() && Objects.nonNull(rdbmsConfig.getSchema())) {
                    tableList = dbManager.getTableAndViewList(db, rdbmsConfig.getSchema());
                } else {
                    tableList = dbManager.getTableAndViewList(db);
                }
            } catch (SQLException e) {
                log.error(e.getMessage());
                throw new BusinessException("获取表列表失败: " + e.getMessage());
            }
        } else if (DSType.Hive.equals(DSType.valueOf(dsType))) {
            DataSourceManagerDomain domain = this.getManagerInfo(dsId);
            if (null == domain) {
                throw new BusinessException("获取表列表失败: " + dsId + "不存在");
            }
            BaseDbManager dbManager = domain.getDbManager();
            try {
                tableList = dbManager.getTableAndViewList(domain.getDbName());
            } catch (SQLException e) {
                log.error(e.getMessage());
                throw new BusinessException("获取表列表失败: " + e.getMessage());
            }
        } else if (DSType.IoTDB.equals(DSType.valueOf(dsType))) {
            DataSourceManagerDomain domain = this.getManagerInfo(dsId);
            if (null == domain) {
                throw new BusinessException("获取表列表失败: " + dsId + "不存在");
            }
            IoTDBSessionManager dbManager = domain.getIoTDBSessionManager();
            try {
                List<DeviceDomain> devices = dbManager.getDeviceList("");
                if (org.apache.commons.collections4.CollectionUtils.isNotEmpty(devices)) {
                    tableList =
                        devices.stream()
                            .map(
                                (d) -> {
                                    TableInfoDomain td = new TableInfoDomain();
                                    td.setTableName(d.getDevice());
                                    td.setPartitionTable(d.isAligned()); // 是否分区->是否是对齐时间序列
                                    return td;
                                })
                            .collect(Collectors.toList());
                }
            } catch (Exception e) {
                log.error(e.getMessage());
                throw new BusinessException("获取表列表失败: " + e.getMessage());
            }
        } else if (DSType.InfluxDB.equals(DSType.valueOf(dsType))) {
            DataSourceManagerDomain domain = this.getManagerInfo(dsId);
            if (null == domain) {
                throw new BusinessException("获取表列表失败: " + dsId + "不存在");
            }
            InfluxDBManager dbManager = domain.getInfluxDBManager();
            try {
                RdbmsConfig rdbmsConfig = dataSourceEntity.getRdbmsConfig();
                tableList = dbManager.getTableAndViewList(rdbmsConfig.getDbName());
            } catch (Exception e) {
                log.error(e.getMessage());
                throw new BusinessException("获取表列表失败: " + e.getMessage());
            }
        }
        String userId = UserContext.getUserId();
        UserDomain userById = userService.getUserById(userId);
        if (Objects.nonNull(userById.getIsAuth()) && userById.getIsAuth().equals(1)) {
            return tableList.stream()
                .sorted(Comparator.comparing(TableInfoDomain::getTableName))
                .collect(Collectors.toList());
        }
        Integer isSystem = dataSourceEntity.getIsSystem();
        if (Objects.nonNull(isSystem) && isSystem.equals(1)) {
            Long projectId = ServletUtils.getProjectId();
            Iterator<TableInfoDomain> iterator = tableList.iterator();
            List<TableInfoBizDomain> tableInfoList =
                tableService.getTableList(projectId, new QueryTableParam());
            Map<String, Long> mapName =
                tableInfoList.stream()
                    .collect(
                        Collectors.toMap(TableInfoBizDomain::getTableName, TableInfoBizDomain::getTblId));
            List<Long> tbInfoIds = tableList.stream().map(e -> e.getTblId()).collect(Collectors.toList());
            Map<Long, String> tblAuthMap = dataAuthService.getTblAuthMap(projectId, userId, tbInfoIds);
            while (iterator.hasNext()) {
                TableInfoDomain next = iterator.next();
                String tableName = next.getTableName();
                Long tblId = mapName.get(tableName);
                if (Objects.isNull(tblId)) {
                    continue;
                }
                String tblAuth = tblAuthMap.get(tblId);
                if (StringUtils.isBlank(tblAuth)) {
                    continue;
                }
                // 来源：无读写权限，过滤掉
                if (RightEnum.NONE.getCode().equals(tblAuth)) {
                    iterator.remove();
                }
            }
        }
        return tableList.stream()
            .sorted(Comparator.comparing(TableInfoDomain::getTableName))
            .collect(Collectors.toList());
    }

    private BaseDbManager getDbManager(
        String dsType, String username, String password, String jdbcLink) {
        Database database = Database.from(dsType);
        DbOptions dbOptions = new DbOptions(database, jdbcLink, username, password);
        return DbManagerFactory.getDbManager(database, dbOptions);
    }

    private BaseDbManager getDbManager(
        String dsType, String username, String password, String jdbcLink, String hiveSiteAddress) {
        Database database = Database.from(dsType);
        DbOptions dbOptions = new DbOptions(database, username, password, jdbcLink, hiveSiteAddress);
        return DbManagerFactory.getDbManager(database, dbOptions);
    }

    private CreateTableParam.FieldInfo toFieldInfo(FieldInfoDomain fieldInfoDomain) {
        CreateTableParam.FieldInfo fieldInfo = new CreateTableParam.FieldInfo();
        BeanUtil.copyProperties(fieldInfoDomain, fieldInfo);
        fieldInfo.setNullable(fieldInfoDomain.isNullable());
        fieldInfo.setPrimaryKey(fieldInfoDomain.isPk());
        return fieldInfo;
    }
}
